package com.shujia.mllib

import org.apache.spark.ml.classification.{LogisticRegression, LogisticRegressionModel}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object Demo2PersonTrain {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("person")
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._

    /**
      * 加载svm格式的数据
      *
      * svm格式的数据由两部组成，一部分是目标值（y）,一部分特征值（x1,x2,x3）
      *
      */
    val data: DataFrame = spark
      .read
      .format("libsvm")
      .load("Spark/data/人体指标.txt")

    // data.printSchema()

    //data.show(false)

    /**
      * 切分训练集和测试集
      *
      * randomSplit: 随机切分，指定切分权重
      */

    val split: Array[DataFrame] = data.randomSplit(Array(0.7, 0.3))

    //训练数据
    val train: DataFrame = split(0)
    //测试数据
    val test: DataFrame = split(1)


    /**
      * 选择算法
      * 我们这里是一个分类问题，可以选择逻辑回归算法
      *
      */
    //构建算法
    val logisticRegression = new LogisticRegression()

    /**
      * 将训练数据带入算法训练模型
      * 再fit 内部会进行不断迭代计算让模型达到最优状态（误差最小）
      *
      */

    val model: LogisticRegressionModel = logisticRegression.fit(train)


    /**
      * 将测试数据带入模型，测试模型准确率
      *
      */
    val testDF: DataFrame = model.transform(test)


    /**
      * 计算准确率
      * 预测对的数量除以总的数量
      */

    testDF
      .select(sum(when($"label" === $"prediction", 1.0).otherwise(0.0)) / count($"label"))
      .show()


    /**
      * 保存模型
      *
      */
    model
      .write
      .overwrite()
      .save("Spark/data/model")

  }

}
